
//Copyright 1997-2009 Syrinx Development, Inc.
//This file is part of the Syrinx Web Application Framework (SWAF).
// == BEGIN LICENSE ==
//
// Licensed under the terms of any of the following licenses at your
// choice:
//
//  - GNU General Public License Version 3 or later (the "GPL")
//    http://www.gnu.org/licenses/gpl.html
//
//  - GNU Lesser General Public License Version 3 or later (the "LGPL")
//    http://www.gnu.org/licenses/lgpl.html
//
//  - Mozilla Public License Version 1.1 or later (the "MPL")
//    http://www.mozilla.org/MPL/MPL-1.1.html
//
// == END LICENSE ==
using System;
using System.Collections;
using System.Diagnostics;
using System.Threading;
using Swaf.BizRule;
using Swaf.Call;
using Swaf.Container;
using Swaf.Logging;

namespace Swaf.Messaging
{
	/// <summary>
	/// Summary description for MessagingListenerGroup.
	/// </summary>
	public class MessagingListenerGroup
	{
		protected IMessagingSystem m_msgSys;
		protected IList m_listeners;
		protected bool m_continueProcessingMessages;
		protected Thread m_listeningThread;
		private LocalDataStoreSlot m_cacheThreadSlot = null;
		protected IList m_expiredCache = new ArrayList();
		/// <summary>
		/// Time to wait in loop after processing a single message from each listener.
		/// </summary>
		protected int m_loopIterationWaitTime = 1000;

		public MessagingListenerGroup(IMessagingSystem msgSys)
		{
			m_msgSys = msgSys;
			m_listeners = new ArrayList();
			m_continueProcessingMessages=false;
			m_listeningThread = new Thread(new ThreadStart(messageProcessingLoop));
			m_listeningThread.Name="MessagingListenerGroup:" + Guid.NewGuid().ToString();
			m_listeningThread.IsBackground = true;
		}

		public IMessagingSystem MessagingSystem {get{return m_msgSys;}}

		public void addListener(IMessagingListener listener)
		{
			lock(m_listeners)
			{
				m_listeners.Add(listener);
			}
		}

		public void start()
		{
			string msgListenerWait = Application.currentApp.resMgr.resolve("$ListenerWaitTime$");
			if (! msgListenerWait.Equals("$ListenerWaitTime$"))
			{
				try
				{
					m_loopIterationWaitTime =int.Parse(msgListenerWait);
				}
				catch(Exception)
				{
					m_loopIterationWaitTime=1000;
				}
			}
			m_continueProcessingMessages = true;
			m_listeningThread.Start();
		}

		public void stop()
		{
			m_continueProcessingMessages = false;
		}
		protected void processExpiredList()
		{
			lock(m_expiredCache)
			{
				foreach(IDisposable content in m_expiredCache)
				{
					try
					{
						content.Dispose();
					}
					catch(Exception e)
					{
						Application.currentApp.Trace.write(TraceLevels.Verbose, "Unable to dispose of expired cache item {0}.  Error was {1}", content.GetType().Name,e.ToString());
					}
				}
				m_expiredCache.Clear();
			}
		}
		private void Worker_NodeExpiring(object contents, ref bool cancel)
		{
			if (contents is IDisposable)
			{
				lock(m_expiredCache)
				{
					m_expiredCache.Add(contents);
				}
			}
		}
		protected void messageProcessingLoop()
		{
			bool results=false;
			//Create a globally available cache map in the thread local storage.
			LocalDataStoreSlot cacheSlot = Thread.GetNamedDataSlot(BasicExecutor.CACHE_THREAD_SLOT_NAME);
			FlexiMap cache = new FlexiMap();
			cache.NodeExpiring +=new ExpirationNotification(Worker_NodeExpiring);
			Thread.SetData(cacheSlot, cache);
			while(m_continueProcessingMessages)
			{
				lock(m_listeners)
				{
					results = false;
					foreach(MessagingListener listener in m_listeners)
						results = results || listener.processSingleMessage();
				}
				//See if any items cached in the thread local cache have expired and need disposing.
				//We do this here to handle any thread safe expiration issues.
				if (m_expiredCache.Count > 0)
				{
					processExpiredList();
				}
				if (!results)
					Thread.Sleep(m_loopIterationWaitTime);
			}
		}
	}
}
